package com.siteseat.crudeq.kernel.impl;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.siteseat.common.foundation.Daemon;
import com.siteseat.common.foundation.PropertyHelper;
import com.siteseat.crudeq.kernel.Message;
import com.siteseat.crudeq.kernel.QueueListener;
import com.siteseat.crudeq.kernel.QueueListenerManager;

/**
 *
 */
public class QueueListenerManagerImpl extends Daemon implements QueueListenerManager {

  private final ReadWriteLock    subscriptionLock   = new ReentrantReadWriteLock();
  
  private Map<Integer,Set<QueueListener>> subscribed = new HashMap<Integer,Set<QueueListener>>();
  
  private BlockingQueue<Message> incoming;
  
  private long waitTimeoutSeconds;

  /**
   * Gets the waitTimeoutSeconds
   *
   * @return the waitTimeoutSeconds
   */
  public long getWaitTimeoutSeconds() {
    return waitTimeoutSeconds;
  }

  /**
   * Sets the waitTimeoutSeconds
   *
   * @param waitTimeoutSeconds the waitTimeoutSeconds to set
   */
  public void setWaitTimeoutSeconds( long waitTimeoutSeconds ) {
    this.waitTimeoutSeconds = waitTimeoutSeconds;
  }

  /**
   * Gets the incoming
   *
   * @return the incoming
   */
  public BlockingQueue<Message> getIncoming() {
    return incoming;
  }

  /**
   * Sets the incoming
   *
   * @param incoming the incoming to set
   */
  public void setIncoming( BlockingQueue<Message> incoming ) {
    this.incoming = incoming;
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  protected void doInit() throws Exception {
    super.doInit();
    
    PropertyHelper.ensureNotNull( log, getIncoming(), "incoming" );
    
    this.getExecutor().execute( this );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doRun() {
    ExecutorService er = getExecutor();
    
    try {
      while( er.isShutdown() == false ) {
        Message msg = incoming.poll( getWaitTimeoutSeconds(), TimeUnit.SECONDS );
        if( msg == null ) {
          continue;
        }
        
        Integer qid = msg.getChunk().getInfo().getQueueId();
        
        Lock k = subscriptionLock.readLock();
        k.lockInterruptibly();
        try {
          Set<QueueListener> listeners = subscribed.get( qid );
          if( listeners != null ) {
            for( QueueListener l: listeners ) {
              l.onMessageCommited( msg );
            }
          }
        } finally {
          k.unlock();
        }
      }
    } catch( InterruptedException e ) {
      log.warn( "Interruppted", e );
      return;
    } 
  }

  /**
   * {@inheritDoc}
   * @throws InterruptedException 
   */
  @Override
  public void subscribe( Integer queueId, QueueListener listener ) throws InterruptedException {    
    Lock k = subscriptionLock.writeLock();
    k.lockInterruptibly();
    try {
      internalSubscribe( this.subscribed, queueId, listener );
    } finally {
      k.unlock();
    }
  }
  
  /**
   *
   * @param queueId
   * @param listener
   */
  protected void internalSubscribe( Map<Integer,Set<QueueListener>> tmp, Integer queueId, QueueListener listener ) {    
    Set<QueueListener> t = tmp.get( queueId );
    if( t == null ) {
      t = new HashSet<QueueListener>();
      tmp.put( queueId, t );
    }
    
    t.add( listener );
  }
  
  /**
   *
   * @param queueId
   * @param listener
   */
  public boolean internalUnsubscribe( Map<Integer,Set<QueueListener>> tmp, Integer queueId, QueueListener listener ) {    
    Set<QueueListener> t = tmp.get( queueId );
    if( t == null ) {
      return false;
    }
    
    return t.remove( listener );
  }
  
  /**
   * {@inheritDoc}
   */
  @Override
  public boolean unsubscribe( Integer queueId, QueueListener listener ) throws InterruptedException {
    Lock k = subscriptionLock.writeLock();
    k.lockInterruptibly();
    try {
      return internalUnsubscribe( this.subscribed, queueId, listener );
    } finally {
      k.unlock();
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void subscribe( Set<Integer> queueIds, QueueListener listener ) throws InterruptedException {
    Lock k = subscriptionLock.writeLock();
    k.lockInterruptibly();
    try {
      Map<Integer,Set<QueueListener>> tmp = new HashMap<Integer,Set<QueueListener>>( this.subscribed );
      
      for( Integer qid: queueIds ) {
        internalSubscribe( tmp, qid, listener );
      }
      
      this.subscribed = tmp;
    } finally {
      k.unlock();
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void unsubscribe( Set<Integer> queueIds, QueueListener listener ) throws InterruptedException {
    Lock k = subscriptionLock.writeLock();
    k.lockInterruptibly();
    try {
      Map<Integer,Set<QueueListener>> tmp = new HashMap<Integer,Set<QueueListener>>( this.subscribed );
      
      for( Integer qid: queueIds ) {
        internalUnsubscribe( tmp, qid, listener );
      }
      
      this.subscribed = tmp;
    } finally {
      k.unlock();
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void onMessageCommited( Message msg ) {
    try {
      incoming.put( msg );
    } catch( InterruptedException e ) {
      log.warn( "Interrupted", e );
    }
  }

}
